package top.cwk.flume.plugin.strategy;

import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @author chenwk <br>
 * @Description: 策略日志拦截器  <br>
 * @date 2020/5/20 5:17 下午  <br>
 * @Copyright tydic.com
 */
public class StrategyInvokeLogInterceptor implements Interceptor {

    //打印日志，便于测试方法的执行顺序
    private static final Logger logger = LoggerFactory.getLogger(StrategyInvokeLogInterceptor.class);

    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        if(event.getBody() != null){
            Map<String, String> headers = event.getHeaders();
            logger.debug("处理拦截数据 -- header : {}",event);
            String reqBody = new String(event.getBody());
            logger.debug("处理拦截数据 -- body : {}",reqBody);
            if(StringUtils.isNotBlank(reqBody)){
                event.setBody(reqBody.getBytes());
                return event;
            }
        }
        return null;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> result = new ArrayList<>();
        Event event;
        for (Event e : events) {
            event = intercept(e);
            if (event != null) {
                result.add(event);
            }
        }
        int i = 1;
        for (Event e : result) {
            logger.info("{} 行数据处理结果：{}" ,i++ ,new String(e.getBody()));
        }
        return result;
    }

    @Override
    public void close() {

    }

    /**
     * 通过该静态内部类来创建自定义对象供flume使用，实现Interceptor.Builder接口，并实现其抽象方法
     */
    public static class Builder implements Interceptor.Builder {
        /**
         * 该方法主要用来返回创建的自定义类拦截器对象
         * @return
         */
        @Override
        public Interceptor build() {
            return new StrategyInvokeLogInterceptor();
        }

        /**
         * 用来接收flume配置自定义拦截器参数
         * @param context 通过该对象可以获取flume配置自定义拦截器的参数
         */
        @Override
        public void configure(Context context) {
            //IP_ADDR = context.getString("ipAddr");
        }

    }
}
